package com.djf.djfcs.listener;

import com.alibaba.fastjson.JSON;
import com.djf.djfcs.canal.CanalDataChangeEvent;
import com.djf.djfcs.canal.CanalDataListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import java.util.Map;

/**
 * 订单表Canal数据监听器
 */
@Slf4j
@Component
public class OrderCanalListener implements CanalDataListener {
    
    @Override
    public void onDataChange(CanalDataChangeEvent event) {
        log.info("订单表数据变更事件: {}.{}, 事件类型: {}", 
                event.getDatabase(), event.getTable(), event.getEventType());
        
        switch (event.getEventType()) {
            case "INSERT":
                handleInsert(event);
                break;
            case "UPDATE":
                handleUpdate(event);
                break;
            case "DELETE":
                handleDelete(event);
                break;
            default:
                log.warn("未知的事件类型: {}", event.getEventType());
        }
    }
    
    /**
     * 处理插入事件
     */
    private void handleInsert(CanalDataChangeEvent event) {
        if (event.getNewData() != null && !event.getNewData().isEmpty()) {
            for (Map<String, Object> row : event.getNewData()) {
                log.info("新增订单数据: {}", JSON.toJSONString(row));
                
                // 这里可以添加业务逻辑，比如：
                // 1. 同步数据到Elasticsearch
                // 2. 发送消息到RabbitMQ
                // 3. 更新缓存
                // 4. 触发其他业务流程
                
                syncToElasticsearch(row, "INSERT");
                sendToRabbitMQ(row, "INSERT");
            }
        }
    }
    
    /**
     * 处理更新事件
     */
    private void handleUpdate(CanalDataChangeEvent event) {
        if (event.getNewData() != null && !event.getNewData().isEmpty()) {
            for (int i = 0; i < event.getNewData().size(); i++) {
                Map<String, Object> newRow = event.getNewData().get(i);
                Map<String, Object> oldRow = null;
                if (event.getOldData() != null && i < event.getOldData().size()) {
                    oldRow = event.getOldData().get(i);
                }
                
                log.info("更新订单数据 - 原数据: {}, 新数据: {}", 
                        oldRow != null ? JSON.toJSONString(oldRow) : "null",
                        JSON.toJSONString(newRow));
                
                syncToElasticsearch(newRow, "UPDATE");
                sendToRabbitMQ(newRow, "UPDATE");
            }
        }
    }
    
    /**
     * 处理删除事件
     */
    private void handleDelete(CanalDataChangeEvent event) {
        if (event.getOldData() != null && !event.getOldData().isEmpty()) {
            for (Map<String, Object> row : event.getOldData()) {
                log.info("删除订单数据: {}", JSON.toJSONString(row));
                
                syncToElasticsearch(row, "DELETE");
                sendToRabbitMQ(row, "DELETE");
            }
        }
    }
    
    /**
     * 同步数据到Elasticsearch
     */
    private void syncToElasticsearch(Map<String, Object> data, String operation) {
        try {
            // 这里可以调用Elasticsearch服务进行数据同步
            log.debug("同步订单数据到Elasticsearch: {} - {}", operation, data.get("id"));
            
            // 示例代码（需要根据实际的ES服务进行调整）:
            // orderEsService.syncOrder(data, operation);
            
        } catch (Exception e) {
            log.error("同步订单数据到Elasticsearch失败", e);
        }
    }
    
    /**
     * 发送消息到RabbitMQ
     */
    private void sendToRabbitMQ(Map<String, Object> data, String operation) {
        try {
            // 这里可以发送消息到RabbitMQ
            log.debug("发送订单变更消息到RabbitMQ: {} - {}", operation, data.get("id"));
            
            // 示例代码（需要根据实际的RabbitMQ配置进行调整）:
            // rabbitTemplate.convertAndSend("order.exchange", "order." + operation.toLowerCase(), data);
            
        } catch (Exception e) {
            log.error("发送订单变更消息到RabbitMQ失败", e);
        }
    }
    
    @Override
    public String getDatabase() {
        return "sharding_order_db"; // 监听分库分表的数据库
    }
    
    @Override
    public String getTable() {
        return null; // 监听所有以order_开头的表，这里返回null监听所有表，在业务逻辑中过滤
    }
    
    @Override
    public String[] getEventTypes() {
        return new String[]{"INSERT", "UPDATE", "DELETE"}; // 监听所有类型的事件
    }
}